package com.xxk.rocket.star.core.producer;

import com.alibaba.rocketmq.client.producer.MQProducer;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.common.message.Message;
import com.xxk.rocket.star.core.builder.ProcessCenter;
import com.xxk.rocket.star.core.context.ProducerContext;
import com.xxk.rocket.star.core.exception.StargateException;
import com.xxk.rocket.star.core.message.IProcessMessageProducer;
import com.xxk.rocket.star.core.dto.ProducerMethodsDTO;
import com.xxk.rocket.star.core.dto.StargateOneWayDTO;
import com.xxk.rocket.star.core.utils.LogUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class ProducerInvocationHandlerImpl implements InvocationHandler {

  private static final Logger log = LoggerFactory.getLogger(ProducerInvocationHandlerImpl.class);
  private final Map<Method, ProducerContext> producerContextMap = new ConcurrentHashMap();

  @Override
  public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
    ProducerContext producerContext = producerContextMap.get(method);
    if (producerContext == null) {
      return objectBaseMethod(method, args);
    }
    MQProducer client = producerContext.getMqProducer();
    if (client == null) {
      throw new StargateException("client is null " + method.getName());
    }
    Message message = new Message();

    for (IProcessMessageProducer producerContextMethod : ProcessCenter
            .getProducersMessageProcess()) {
      log.debug(
              "IProcessMessageProducer for " + method.getName() + ":" + producerContextMethod.getClass()
                      .getName() + " ing");
      int result = producerContextMethod.processProducer(producerContext, message, args);
      if (result == IProcessMessageProducer.BREAK) {
        log.debug(
                "IProcessMessageProducer for " + method.getName() + ":" + producerContextMethod.getClass()
                        .getName() + " BREAK");
        return null;
      }
      if (result == IProcessMessageProducer.SEND) {
        log.debug(
                "IProcessMessageProducer for " + method.getName() + ":" + producerContextMethod.getClass()
                        .getName() + " SEND");
        break;
      }
    }

    StargateOneWayDTO stargateOneWayDTO = producerContext.getDTO(StargateOneWayDTO.class);
    ProducerMethodsDTO producerMethodsDTO = producerContext.getDTO(ProducerMethodsDTO.class);

    LogUtils.messageLog(message);
    if (stargateOneWayDTO.isOneWay()) {
      log.debug("send message for " + method.getName() + ":sendOneway");
      client.sendOneway(message);
      return null;
    }
    if (producerMethodsDTO.getSendCallbackIndex() > 0) {
      log.debug("send message for " + method.getName() + ":SendCallback");
      client.send(message, (SendCallback) args[producerMethodsDTO.getSendCallbackIndex()]);
      return null;
    }
    log.debug("send message for " + method.getName() + ":send");
    return client.send(message);
  }

  private Object objectBaseMethod(Method method, Object[] args)
          throws Exception {
    if ("toString".equals(method.getName()) && method.getParameterTypes().length == 0 &&
            method.getReturnType().equals(String.class)) {
      return "StargateProducerProxy";
    }
    return method.invoke(this, args);
  }

  public void putContextMap(Method method, ProducerContext producerContext) {
    this.producerContextMap.put(method, producerContext);
  }
}
